package com.practice.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.practice.gmall.realtime.bean.TableProcess;
import com.practice.gmall.realtime.common.Constant;
import com.practice.gmall.realtime.sink.PhoenixSink;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Properties;
import java.util.stream.Collectors;
@Slf4j
public class FlinkSinkUtil {
    public static PhoenixSink getPhoenixSink() {
        return new PhoenixSink();
    }


    public static SinkFunction<String> getKafkaSink(String topic) {

        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Constant.KAFKA_BROKERS);
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, String.valueOf(15 * 60 * 1000L));

        return new FlinkKafkaProducer<String>(
                "default",
                new KafkaSerializationSchema<String>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long aLong) {
                        return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
                    }
                },
                props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }

    public static SinkFunction<Tuple2<JSONObject, TableProcess>> kafkaTuple2SinkFunc() {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms", 15 * 60 * 1000);
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
                "default",
                new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> t,
                                                                    @Nullable Long timestamp) {
                        String topic = t.f1.getSinkTable();
                        JSONObject data = t.f0;

                        return new ProducerRecord<>(topic, data.toJSONString().getBytes(StandardCharsets.UTF_8));
                    }
                },
                props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }

    /**
     * @param table  写入到clickhouse的表名
     * @param tClass Flink流中的数据的封装形式
     * @Description: 这是一个Flink自定义Sink函数，用于将Flink流中的数据输出到Clickhouse
     * @return: org.apache.flink.streaming.api.functions.sink.SinkFunction<T>
     * @Author: lzy
     * @Date:2023-02-17 - 16:10
     */
    public static <T> SinkFunction<T> getClickhouseSink(String table, Class<T> tClass) {

        // todo 如何区分哪些代码是在客户端运行，哪些代码是在服务端运行？
        StringBuilder sql = new StringBuilder();
        Field[] fields = tClass.getDeclaredFields();
        // insert into table_name(field1, field2, field3, field4) values(?,?,?,?)

        String columns = Arrays.stream(fields)
                .map(Field::getName)
                .map(CommonUtil::smallCamelCaseToUnderscore)
                .collect(Collectors.joining(","));

        String placerHolder = columns.replaceAll("[^,]+","?");

        sql.append("insert into ")
                .append(table)
                .append("(")
                .append(columns)
                .append(") values (")
                .append(placerHolder)
                .append(")");

       // System.out.println(sql);// test sql statement formatter

        return JdbcSink.sink(sql.toString(),
                new JdbcStatementBuilder<T>()  {
                    @Override
                    public void accept(PreparedStatement ps, T t) {
                        Field[] fields = t.getClass().getDeclaredFields();
                        for (int i = 0; i < fields.length; i++) {
                            fields[i].setAccessible(true);
                            try {
                                ps.setObject(i+1, fields[i].get(t));
                            } catch (SQLException e) {
                                log.error("can't set value to " + fields[i].getName());
                                throw new RuntimeException(e);
                            } catch (IllegalAccessException e) {
                                log.error("can't access to field : " + fields[i].getName());
                                throw new RuntimeException(e);
                            }
                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(10 * 1000 * 1000)
                        .withBatchIntervalMs(5000L)
                        .withMaxRetries(4)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(Constant.CLICKHOUSE_URL)
                        .withDriverName(Constant.CLICKHOUSE_DRIVER)
                        .withUsername(Constant.CLICKHOUSE_USER)
                        .withPassword(Constant.CLICKHOUSE_PASSWORD)
                        .withConnectionCheckTimeoutSeconds(5)
                        .build()
        );
    }
}
